package com.asap.demo.sourcefunc;

import com.alibaba.fastjson.JSON;
import com.asap.demo.model.RuleBean;
import com.asap.rule.engine.InferenceEngine;
import com.asap.rule.orm.DbFetcher;
import com.asap.rule.util.RuleReader;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.source.RichSourceFunction;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.sql.*;
import java.util.ArrayList;
import java.util.List;

public class MysqlSourceFunction extends RichSourceFunction<InferenceEngine> {

	private static final Logger logger = LoggerFactory.getLogger(MysqlSourceFunction.class);


	DbFetcher dbFetcher;

	public MysqlSourceFunction(DbFetcher dbFetcher) {
		this.dbFetcher = dbFetcher;
	}
	@Override
	public void run(SourceContext<InferenceEngine> ctx) throws Exception {
		logger.info("get config ..............");
		InferenceEngine engine = InferenceEngine.compile(RuleReader.parseRules(RuleReader.readRules(dbFetcher)));
		ctx.collect(engine);

		while (true) {
			if (dbFetcher.getConfigUpdateLog() == 1) {
				logger.info("config update..............");
				engine = InferenceEngine.compile(RuleReader.parseRules(RuleReader.readRules(dbFetcher)));
				ctx.collect(engine);
				dbFetcher.updateConfigUpdateLog();
			}
			Thread.sleep(10000);
		}


	}

	@Override
	public void cancel() {

	}

	@Override
	public void close() throws Exception {
		super.close();

	}

	@Override
	public void open(Configuration parameters) throws Exception {
		super.open(parameters);

	}

	@Override
	protected Object clone() throws CloneNotSupportedException {
		return super.clone();
	}



}
